package com.raft.core.rpc.nio;

import java.io.ByteArrayOutputStream;
import java.io.IOException;

import com.alibaba.com.caucho.hessian.io.HessianOutput;
import com.raft.core.node.NodeId;
import com.raft.core.rpc.message.AppendEntriesResult;
import com.raft.core.rpc.message.AppendEntriesRpc;
import com.raft.core.rpc.message.InstallSnapshotResult;
import com.raft.core.rpc.message.InstallSnapshotRpc;
import com.raft.core.rpc.message.MessageConstants;
import com.raft.core.rpc.message.RequestVoteResult;
import com.raft.core.rpc.message.RequestVoteRpc;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

/**编码响应的类
 * @author zhangsulei
 *
 */
public class Encoder extends MessageToByteEncoder<Object> {

    @Override
    protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
        if (msg instanceof NodeId) {
        	// 这个处理比较简单，因为value就是一个字符串
            this.writeMessage(out, MessageConstants.MSG_TYPE_NODE_ID, ((NodeId) msg).getValue().getBytes());
        } else if (msg instanceof RequestVoteRpc) {
            RequestVoteRpc rpc = (RequestVoteRpc) msg;
            this.writeMessage(out, MessageConstants.MSG_TYPE_REQUEST_VOTE_RPC, rpc);
        } else if (msg instanceof RequestVoteResult) {
            RequestVoteResult result = (RequestVoteResult) msg;
            this.writeMessage(out, MessageConstants.MSG_TYPE_REQUEST_VOTE_RESULT, result);
        } else if (msg instanceof AppendEntriesRpc) {
            AppendEntriesRpc rpc = (AppendEntriesRpc) msg;
            this.writeMessage(out, MessageConstants.MSG_TYPE_APPEND_ENTRIES_RPC, rpc);
        } else if (msg instanceof AppendEntriesResult) {
            AppendEntriesResult result = (AppendEntriesResult) msg;
            this.writeMessage(out, MessageConstants.MSG_TYPE_APPEND_ENTRIES_RESULT, result);
        } else if (msg instanceof InstallSnapshotRpc) {
            InstallSnapshotRpc rpc = (InstallSnapshotRpc) msg;
            this.writeMessage(out, MessageConstants.MSG_TYPE_INSTALL_SNAPSHOT_PRC, rpc);
        } else if (msg instanceof InstallSnapshotResult) {
            InstallSnapshotResult result = (InstallSnapshotResult) msg;
            this.writeMessage(out, MessageConstants.MSG_TYPE_INSTALL_SNAPSHOT_RESULT, result);
        }
    }

    private void writeMessage(ByteBuf out, int messageType, Object message) throws IOException {
        ByteArrayOutputStream byteOutput = new ByteArrayOutputStream();
        HessianOutput hessianOutput = new HessianOutput(byteOutput);
        hessianOutput.writeObject(message);
        // 4(messageType) + 4(bytes.length) + VAR(bytes)
        out.writeInt(messageType);
        this.writeBytes(out, byteOutput.toByteArray());
    }

    private void writeMessage(ByteBuf out, int messageType, byte[] bytes) {
        // 4(messageType) + 4(bytes.length) + VAR(bytes)
        out.writeInt(messageType);
        this.writeBytes(out, bytes);
    }

    private void writeBytes(ByteBuf out, byte[] bytes) {
        out.writeInt(bytes.length);
        out.writeBytes(bytes);
    }

}
